import json
from time import sleep
from confluent_kafka import Consumer
from confluent_kafka.cimpl import KafkaError
from Queen.config.queen_config import queen_consumer_config, slave_to_queen_topic
from logger_config import logger_factory

logger = logger_factory()


def queen_consume():
    consumer = Consumer(queen_consumer_config)
    consumer.subscribe([slave_to_queen_topic])
    logger.info('')
    try:
        while True:
            msg = consumer.poll(timeout=1.0)
            if msg is None:
                logger.info('')
                sleep(3)
                continue
            if msg.error():
                if msg.error().code() == KafkaError.PARTITION_EOF:
                    logger.error('%% %s [%d] reached end of offset %d\n' %(msg.topic(),msg.partition(),msg.offset()))
            else:
                info_dict = json.loads(msg.value().decode('utf-8'))
                mission_name = info_dict['mission']
                lot_id = info_dict['LOTID']
                slave_id = info_dict['SLAVEID']
    except:
        logger.error('')
    finally:
        consumer.close()


